package com.jonah.demo.hbase.thrift;

import com.jonah.demo.hbase.thrift.config.HBaseThriftConfig;
import org.apache.hadoop.hbase.thrift.generated.Hbase;
import org.apache.hadoop.hbase.thrift.generated.TCell;
import org.apache.hadoop.hbase.thrift.generated.TRowResult;
import org.apache.hadoop.hbase.thrift.generated.TScan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ThriftScanDemo {
    public static void main(String[] args) throws TException {
        HBaseThriftConfig instant = HBaseThriftConfig.getInstant();
        //
        String startRow = instant.getStartRow();

        String endRow = instant.getEndRow();


        String thriftHost = instant.getThriftHost();
        Integer thriftPortInt = instant.getThriftPortInt();
        String tableName = instant.getTableName();
        TTransport transport = new TSocket(thriftHost, thriftPortInt, 5000);
        TProtocol protocol = new TBinaryProtocol(transport);
        Hbase.Client client = new Hbase.Client(protocol);
        transport.open();
        ByteBuffer table = ByteBuffer.wrap(tableName.getBytes());
        TScan scan = new TScan();
//        scan.setStartRow(ByteBuffer.wrap(Bytes.toBytes(startRow)));
//        scan.setStopRow(ByteBuffer.wrap(Bytes.toBytes(endRow)));
        Map<ByteBuffer, ByteBuffer> attributes = new HashMap<ByteBuffer, ByteBuffer>();

        int i = client.scannerOpenWithScan(table, scan, attributes);
//        List<TRowResult> tRowResults = client.scannerGet(i);
        List<TRowResult> tRowResults  = client.scannerGetList(i,100);
        for (TRowResult tRowResult : tRowResults) {
            byte[] row = tRowResult.getRow();
            Map<ByteBuffer, TCell> columns = tRowResult.getColumns();
            for (Map.Entry<ByteBuffer, TCell> byteBufferTCellEntry : columns.entrySet()) {
                ByteBuffer key = byteBufferTCellEntry.getKey();
                TCell value = byteBufferTCellEntry.getValue();
                System.out.println(utf8(value.getValue()));
            }

        }


        while (scan.isReversed()){
            byte[] startRow1 = scan.getStartRow();
            System.out.println(Bytes.toString(startRow1));
        }


        System.out.println(attributes);



//        for (long index = startRowLong; index < endRowLong; index++) {
//
//
//            String phone = String.valueOf(index);
//            String md5 = MD5Utils.getMD5Code(phone);
//            ByteBuffer row = ByteBuffer.wrap(md5.getBytes());
//            ByteBuffer data = ByteBuffer.wrap("data".getBytes());
//            List<TCell> rowOrBefore = client.getRowOrBefore(table, row, data);
//            if (rowOrBefore == null) {
//                System.out.println("rowOrBefore is null");
//            } else {
//                for (TCell tCell : rowOrBefore) {
//
//                    System.out.println("tcell value = " + utf8(tCell.getValue()));
//                }
//
//
//            }
//
//
//        }

    }

    private static String utf8(byte[] buf) {
        try {
            CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();
            return decoder.decode(ByteBuffer.wrap(buf)).toString();
        } catch (CharacterCodingException e) {
            return "[INVALID UTF-8]";
        }
    }

    private static byte[] bytes(String s) {
        try {
            return s.getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            return null;
        }
    }
}
